package com.ds.infrastructure.hyperspace.container.endpoint.entry;

import com.ds.infrastructure.hyperspace.common.Endpoint;
import com.ds.infrastructure.hyperspace.common.EndpointProtocolEnum;
import com.ds.infrastructure.hyperspace.common.config.EntryEndpointConfig;
import com.ds.infrastructure.hyperspace.common.config.EntryNodeConfig;
import com.ds.infrastructure.hyperspace.container.endpoint.EndpointInitException;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import java.util.List;

import static com.ds.infrastructure.hyperspace.container.internal.NettyContainer.OBJECT_MAPPER;
import static io.netty.channel.ChannelOption.*;

/**
 * @author  Weishaoying
 */
@Slf4j
public class NettyEntryEndpoint implements Endpoint<EntryEndpointConfig> {

    private final NettyContainer nettyContainer;

    private volatile EntryEndpointConfig entryEndpointConfig;

    public NettyEntryEndpoint(EntryEndpointConfig entryEndpointConfig, NettyContainer nettyContainer) throws Exception {
        this.nettyContainer = nettyContainer;
        this.entryEndpointConfig = entryEndpointConfig;
        init();
    }

    @Override
    public void init() throws Exception {
        List<EntryNodeConfig> nodeConfigList = entryEndpointConfig.getNodes();
        if (CollectionUtils.isEmpty(nodeConfigList)) {
            log.warn("Entry endpoint nodes not found!");
            return;
        }
        for (EntryNodeConfig entryNodeConfig : nodeConfigList) {
            listen(entryNodeConfig);
        }
    }

    @Override
    public void reload(EntryEndpointConfig entryEndpointConfig) throws Exception {
        this.entryEndpointConfig = entryEndpointConfig;
        init();
    }

    @Override
    public EntryEndpointConfig getEndpointConfig() {
        return entryEndpointConfig;
    }

    @Override
    public String getNamespace() {
        return entryEndpointConfig.getNamespace();
    }

    private void listen(EntryNodeConfig entryNodeConfig) throws Exception {
        try {
            if (entryNodeConfig.getProtocolConfig() == null) {
                entryNodeConfig.setProtocolConfig(nettyContainer.getProtocolConfig());
            }
            EndpointProtocolEnum existsEntryEndpointProtocol =
                    nettyContainer.getEntryEndpointProtocol(
                            entryNodeConfig.getHost(), entryNodeConfig.getPort());
            if (existsEntryEndpointProtocol != null) {
                if (existsEntryEndpointProtocol != entryNodeConfig.getProtocol()) {
                    throw new EndpointInitException("Init node " +
                            OBJECT_MAPPER.writeValueAsString(entryNodeConfig) +
                            " failed, another protocol bind this port!");
                }
                return;
            }
            ServerBootstrap serverBootstrap = new ServerBootstrap()
                    .group(nettyContainer.getEntryAcceptorEventLoopGroup(),
                            nettyContainer.getEntryWorkerEventLoopGroup())
                    .channel(nettyContainer.getNettyTransportFactoryDelegate().serverSocketChannelClass())
                    .option(SO_BACKLOG, 102400)
                    .option(SO_REUSEADDR, true)
                    .childOption(SO_KEEPALIVE, true)
                    .childOption(TCP_NODELAY, true)
                    .childOption(ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .childHandler(new NettyEntryEndpointChannelInitializer(nettyContainer, entryNodeConfig));
            serverBootstrap.bind(entryNodeConfig.getHost(), entryNodeConfig.getPort()).sync();
            nettyContainer.putEntryEndpointProtocol(entryNodeConfig.getHost(),
                    entryNodeConfig.getPort(), entryNodeConfig.getProtocol());
            log.info("Entry endpoint bind successfully on [{}:{}].",
                    entryNodeConfig.getHost(), entryNodeConfig.getPort());
        } catch (Exception e) {
            log.error("Bind entry endpoint failed! host {}, port {}, protocol {}.",
                    entryNodeConfig.getHost(), entryNodeConfig.getPort(), entryNodeConfig.getProtocol());
            throw  new EndpointInitException("Bind entry endpoint node failed! node info [" +
                    entryNodeConfig.getHost() + ":" + entryNodeConfig.getPort() + "]");
        }
    }
}
